package com.qzuser.sendmessagebasic.server.queue;

import com.qzuser.sendmessagebasic.common.exception.ConsumeException;
import com.qzuser.sendmessagebasic.common.exception.MessageFileException;
import com.qzuser.sendmessagebasic.server.VirtualHost;
import com.qzuser.sendmessagebasic.server.message.Message;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

public class ConsumeCenter {
    private final VirtualHost virtualHost;
    private final BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();
    private final ExecutorService workerPool = Executors.newFixedThreadPool(4);

    public ConsumeCenter(VirtualHost virtualHost) {
        this.virtualHost = virtualHost;
        Thread scanThread = new Thread(() -> {
            try {
                while (true) {
                    String queueName = tokenQueue.take();
                    Queue queue = virtualHost.getMemoryCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new ConsumeException("[ConsumeCenter] 取令牌时，当前队列不存在：" + queueName);
                    }
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        scanThread.setDaemon(true);
        scanThread.start();
    }

    private void consumeMessage(Queue queue) {
        ConsumerEntity luckyDog = queue.getConsumer();
        if (luckyDog == null) {
            return;
        }
        Message message = virtualHost.getMemoryCenter().pushMessage(queue.getName());
        if (message == null) {
            return;
        }
        workerPool.submit(() -> {
            try {
                virtualHost.getMemoryCenter().addWaitAckMsg(queue.getName(), message);
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getProperties(), message.getBody());
                if (luckyDog.isAutoAck()) {
                    if (message.getDeliverMode() == 1) {
                        virtualHost.getDiskCenter().deleteMessage(queue, message, virtualHost.getMemoryCenter());
                    }
                    virtualHost.getMemoryCenter().deleteMessage(message.getMessageId());
                    virtualHost.getMemoryCenter().deleteWaitAckMsg(queue.getName(), message.getMessageId());
                }
                System.out.println("[ConsumeCenter] 消息消费成功：" + queue.getName() + " -> " + message.getMessageId());
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    public void notifyConsume(String name) throws InterruptedException {
        tokenQueue.put(name);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws ConsumeException {
        Queue queue = virtualHost.getMemoryCenter().getQueue(queueName);
        if (queue == null) {
            throw new ConsumeException("[ConsumeCenter] 队列不存在无法添加消费者：" + queueName);
        }
        ConsumerEntity entity = new ConsumerEntity(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumer(entity);
            int msgCount = virtualHost.getMemoryCenter().getQueueMsgCount(queueName);
            for (int i = 0; i < msgCount; i++) {
                consumeMessage(queue);
            }
        }
    }
}
